package cn.doitedu.sql;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import java.text.ParseException;
import java.util.Date;

public class Demo12_UserDefinedFunction {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.createTemporaryFunction("split",StringSplit.class);

        // 数组从1开始取
        //tenv.executeSql("select split('a,b,c,dd,ee', ',' ),split('a,b,c,dd,ee', ',' )[1],split('a,b,c,dd,ee', ',' )[3]  ").print();
        //tenv.executeSql("select array[3,5,7][1]").print();  // 3


        tenv.createTemporaryFunction("time_floor",TimeTruncate.class);
        //tenv.executeSql("select time_floor(1713148144000, 30)").print();


        tenv.createTemporaryFunction("time_str_floor",TimeStrTruncate.class);
        tenv.executeSql("select time_str_floor('2024-04-15 10:22:30.080', 30,'yyyy-MM-dd HH:mm:ss.SSS')").print();


    }


    /**
     * 字符串切割
     */
    public static class StringSplit extends ScalarFunction{

        public String[] eval(String str, String sep){

            return str.split(sep);

        }
    }



    public static class TimeTruncate extends ScalarFunction{

        public String eval(Long timePoint, int interval_minute ){

            // 2024-04-15 10:22:30.080 ==> 5分钟取整 ==> 2024-04-15 10:20:00.000
            // 2024-04-15 10:26:30.080 ==> 5分钟取整 ==> 2024-04-15 10:25:00.000
            // 2024-04-15 10:37:30.080 ==> 5分钟取整 ==> 2024-04-15 10:25:00.000

            long ms = interval_minute*60*1000L;

            long tmp = (timePoint/ms)*ms;

            return DateFormatUtils.format(tmp,"yyyy-MM-dd HH:mm:ss.SSS");


        }


    }



    // TODO ,接收  2024-04-15 10:31.20.080 字符串，和一个取整长度，对它进行取整
    public static class TimeStrTruncate extends ScalarFunction{

        public String eval(String timePointStr, int interval_minute ,String pattern) throws ParseException {

            // 2024-04-15 10:22:30.080 ==> 5分钟取整 ==> 2024-04-15 10:20:00.000
            // 2024-04-15 10:26:30.080 ==> 5分钟取整 ==> 2024-04-15 10:25:00.000
            // 2024-04-15 10:37:30.080 ==> 5分钟取整 ==> 2024-04-15 10:25:00.000

            long timePoint = DateUtils.parseDate(timePointStr, pattern).getTime();

            long ms = interval_minute*60*1000L;

            long tmp = (timePoint/ms)*ms;

            return DateFormatUtils.format(tmp,"yyyy-MM-dd HH:mm:ss.SSS");


        }


    }


}
